package com.oowoo.mq2es.listener;

import com.alibaba.fastjson.JSON;
import com.oowoo.mq2es.dao.BlogTagRepository;
import com.oowoo.mq2es.entity.BinlogDto;
import com.oowoo.mq2es.entity.BlogTagEntity;
import com.oowoo.mq2es.util.TimeUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.List;
/*
 * @Author ZhangBin
 * @Date 2019/12/6 20:37 
 * @Description: 将kafka中的数据增量更新到ES中
 * 
 **/

@Slf4j
@Component
public class KafkaConsumer {
    @Autowired
    private BlogTagRepository blogTagRepository;

    public static final String UPDATE = "update";
    public static final String INSERT = "insert";
    public static final String DELETE = "delete";

    @KafkaListener(topics = {"binlog_blogcore"})
    public void listen(ConsumerRecord<?, String> record) {
        if(record == null || record.value() == null){
            return;
        }
        log.info("kafka收到消息---->{}",record.value());
        BinlogDto binlogDto = JSON.parseObject(record.value(), BinlogDto.class);
        String event = binlogDto.getEvent();
        Serializable value = binlogDto.getValue();

        //将数据存入ES
        data2Es(event,value);
    }

    public void data2Es(String event,Serializable value){
        String[] eventArr = event.split("\\.");//数据库名.表名.增删改查类型:  oowoo_back.blog_tag.update
        if("blog_tag".equals(eventArr[1])){
            BlogTagEntity blogTagEntity = value2BlogTag(value);
            if(UPDATE.equals(eventArr[2])){//更新操作
                blogTagRepository.save(blogTagEntity);//局部更新，通过id更新数据，有数据就更新，没有数据则会插入数据
            }else if(INSERT.equals(eventArr[2])){//插入操作
                blogTagRepository.save(blogTagEntity);
            }else if(DELETE.equals(eventArr[2])){//删除操作
                blogTagRepository.delete(blogTagEntity);
            }
        }else if("blog_core".equals(eventArr[1])){
            Object obj = value2BlogCore(value);
        }
    }

    //将数组对象转化为BlogTagEntity
    public BlogTagEntity value2BlogTag(Serializable value){
        List<Object> list = (List) value;
        BlogTagEntity blogTagEntity = new BlogTagEntity();
        blogTagEntity.setId((Integer)list.get(0));
        blogTagEntity.setTagCode((String)list.get(1));
        blogTagEntity.setTagName((String)list.get(2));
        if(list.get(3)!=null && list.get(3) instanceof Long){
            blogTagEntity.setCreateTime(TimeUtil.getDateTimeOfTimestamp((long)list.get(3)));
        }else{
            blogTagEntity.setCreateTime((LocalDateTime)list.get(3));
        }
        if(list.get(4)!=null && list.get(4) instanceof Long){
            blogTagEntity.setUpdateTime(TimeUtil.getDateTimeOfTimestamp((long)list.get(4)));
        }else {
            blogTagEntity.setUpdateTime((LocalDateTime)list.get(4));
        }
        return blogTagEntity;
    }
    //将数组对象转化为一个指定的实体类
    public Object value2BlogCore(Serializable value){
        return new Object();
    }



}
